package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"time"
)

type KeyValue struct {
	Key   string
	Value string
}

type SortedKey []KeyValue

func (k SortedKey) Len() int {
	return len(k)
}

func (k SortedKey) Swap(i, j int) {
	k[i], k[j] = k[j], k[i]
}

func (k SortedKey) Less(i, j int) bool {
	return k[i].Key < k[j].Key
}
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	keepFlag := true
	for keepFlag {
		task := GetTask()
		switch task.TaskType {
		case MapTask:
			{
				DoMapTask(mapf, &task)
				callDone(&task)
			}
		case ReduceTask:
			{
				DoReduceTask(reducef, &task)
				callDone(&task)
			}
		case WaitingTask:
			{
				time.Sleep(time.Second * 5)
			}
		case ExitTask:
			{
				keepFlag = false
			}
		}
	}
}

func DoReduceTask(reducef func(string, []string) string, t *Task) {
	reduceFileNum := t.TaskId
	intermediate := shuffle(t.FileSlice)
	dir, _ := os.Getwd()
	tempFile, err := ioutil.TempFile(dir, "mr-tmp-*")
	if err != nil {
		log.Fatal("Failed to create tmp file", err)
	}
	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		var values []string
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)
		fmt.Fprintf(tempFile, "%v %v\n", intermediate[i].Key, output)
		i = j
	}
	tempFile.Close()
	fn := fmt.Sprintf("mr-out-%d", reduceFileNum)
	os.Rename(tempFile.Name(), fn)
}

func shuffle(files []string) []KeyValue {
	var kva []KeyValue
	for _, filepath := range files {
		file, _ := os.Create(filepath)
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
		file.Close()
	}
	sort.Sort(SortedKey(kva))
	return kva
}

func callDone(t *Task) Task {
	args := t
	reply := Task{}
	ok := call("Master.MarkFinished", &args, &reply)
	if ok {
		fmt.Printf("worker finished: taskId [ %d ]", args.TaskId)
	} else {
		fmt.Println("call failed")
	}
	return reply
}

func DoMapTask(mapf func(string, string) []KeyValue, response *Task) {
	var intermediate []KeyValue
	filename := response.FileSlice[0]
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("cannot open file: %v , error: %v", filename, err)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read file: %v , error: %v", filename, err)
	}
	file.Close()

	// Mapf 返回一组KV结构体Slice
	intermediate = mapf(filename, string(content))

	reduceNum := response.ReduceNum
	HashedKV := make([][]KeyValue, reduceNum)
	for _, kv := range intermediate {
		HashedKV[ihash(kv.Key)%reduceNum] = append(HashedKV[ihash(kv.Key)%reduceNum], kv)
	}

	// 将结果输出到中间文件中
	for i := 0; i < reduceNum; i++ {
		oname := fmt.Sprintf("mr-tmp-%d-%d", strconv.Itoa(response.TaskId), strconv.Itoa(i))
		ofile, _ := os.Create(oname)
		enc := json.NewEncoder(ofile)
		for _, kv := range HashedKV[i] {
			enc.Encode(kv)
		}
		ofile.Close()
	}

}

func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

func GetTask() Task {
	args := TaskArgs{}
	reply := Task{}
	ok := call("Master.PollTask", &args, &reply)
	if ok {
		fmt.Println("Task:", reply)
	} else {
		fmt.Println("call failed")
	}
	return reply
}

func call(rpcname string, args interface{}, reply interface{}) bool {
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()
	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}
	fmt.Println(err)
	return false
}
